package com.xinqi.common.rocketmq.event;

import com.xinqi.common.core.utils.SpringUtils;
import com.xinqi.common.rocketmq.dto.MqMessageDTO;
import com.xinqi.common.rocketmq.dto.RocketMqBean;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.apis.ClientServiceProvider;
import org.apache.rocketmq.client.apis.message.Message;
import org.apache.rocketmq.client.apis.producer.Producer;
import org.apache.rocketmq.client.apis.producer.SendReceipt;

import java.nio.charset.StandardCharsets;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 发布消息
 *
 * @author dzy
 */
@Slf4j
public class ProducerEvent {

    private static final RocketMqBean rocketMqBean = SpringUtils.getBean(RocketMqBean.class);

    /**
     * 普通消息
     * <p>
     * 普通消息一般应用于微服务解耦、事件驱动、数据集成等场景，
     * 这些场景大多数要求数据传输通道具有可靠传输的能力，且对消息的处理时机、处理顺序没有特别要求。
     * <p>
     * 同步发送
     *
     * @param mqMessage
     */
    public static void producerNormal(MqMessageDTO mqMessage) {
        if (Objects.isNull(mqMessage)) {
            return;
        }
        ClientServiceProvider provider = ClientServiceProvider.loadService();
        Producer producer = null;
        try {
            producer = ProducerSingleton.getInstance(rocketMqBean.getTopic());
        } catch (ClientException e) {
            log.info("RocketMq连接失败 => {}", e.getMessage());
            return;
        }
        // Define your message body.
        byte[] body = mqMessage.getMessageBody().getBytes(StandardCharsets.UTF_8);
        Message message = provider.newMessageBuilder()
            .setTopic(rocketMqBean.getTopic())
            .setTag(mqMessage.getMessageTag())
            .setKeys(mqMessage.getMessageKey())
            .setBody(body)
            .build();
        try {
            SendReceipt sendReceipt = producer.send(message);
            log.info("RocketMq发送消息成功 => {}", sendReceipt.getMessageId());
        } catch (Throwable t) {
            log.info("RocketMq发送消息失败 => {}", t.getMessage());
        }
    }

    /**
     * 普通消息
     * <p>
     * 普通消息一般应用于微服务解耦、事件驱动、数据集成等场景，
     * 这些场景大多数要求数据传输通道具有可靠传输的能力，且对消息的处理时机、处理顺序没有特别要求。
     * <p>
     * 异步发送
     *
     * @param mqMessage
     */
    public static void asyncProducer(MqMessageDTO mqMessage) {
        if (Objects.isNull(mqMessage)) {
            return;
        }
        ClientServiceProvider provider = ClientServiceProvider.loadService();
        Producer producer = null;
        try {
            producer = ProducerSingleton.getInstance(rocketMqBean.getTopic());
        } catch (ClientException e) {
            log.info("RocketMq连接失败 => {}", e.getMessage());
            return;
        }
        byte[] body = mqMessage.getMessageBody().getBytes(StandardCharsets.UTF_8);
        Message message = provider.newMessageBuilder()
            .setTopic(rocketMqBean.getTopic())
            .setTag(mqMessage.getMessageTag())
            .setKeys(mqMessage.getMessageKey())
            .setBody(body)
            .build();
        CompletableFuture<SendReceipt> future = producer.sendAsync(message);
        ExecutorService sendCallbackExecutor = Executors.newCachedThreadPool();
        future.whenCompleteAsync((sendReceipt, t) -> {
            if (null != t) {
                log.info("RocketMq发送消息失败 => {}", t.getMessage());
                return;
            }
            log.info("RocketMq发送消息成功 => {}", sendReceipt.getMessageId());
        }, sendCallbackExecutor);
    }

    /**
     * 顺序消息
     * <p>
     * 消费者按照发送消息的先后顺序获取消息，从而实现业务场景中的顺序处理
     *
     * @param mqMessage
     */
    public static void producerFifoMessage(MqMessageDTO mqMessage) {
        if (Objects.isNull(mqMessage)) {
            return;
        }
        ClientServiceProvider provider = ClientServiceProvider.loadService();
        Producer producer = null;
        try {
            producer = ProducerSingleton.getInstance(rocketMqBean.getTopic());
        } catch (ClientException e) {
            log.info("RocketMq连接失败 => {}", e.getMessage());
            return;
        }
        byte[] body = mqMessage.getMessageBody().getBytes(StandardCharsets.UTF_8);
        Message message = provider.newMessageBuilder()
            .setTopic(rocketMqBean.getTopic())
            .setTag(mqMessage.getMessageTag())
            .setKeys(mqMessage.getMessageKey())
            .setBody(body)
            .build();
        try {
            SendReceipt sendReceipt = producer.send(message);
            log.info("RocketMq发送消息成功 => {}", sendReceipt.getMessageId());
        } catch (Throwable t) {
            try {
                log.info("RocketMq发送消息失败 => {}", t.getMessage());
                //发送失败的消息一秒钟后重启发送
                Thread.sleep(1000L);
            } catch (InterruptedException e) {

            }
            producerFifoMessage(mqMessage);
        }
    }


}
